/***************************************************************************

MapCG: MapReduce Framework for CPU & GPU

Copyright (C) 2010, Chuntao HONG (Chuntao.Hong@gmail.com).

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program.  If not, see <http://www.gnu.org/licenses/>.

**************************************************************************/

#include <iostream>
#include <fstream>
#include <stdlib.h>
#include <string>
#include "time.h"
#include "math.h"
#include <vector>
#include <assert.h>
using namespace std;

#include "HMM/HMMMain.h"
#include "HMM/UtilLib/HMMCommUtil.h"
#include "HMM/UtilLib/HMMCommUtilTemplate.h"
#include "DS.h"


#define MAX_WORD_SIZE	255

int main(int argc, char **argv){   
	//----------------------------------------------
	//get parameters
	//----------------------------------------------
	string relation1Filename;
	string relation2Filename;

	// check input options are present
	if(!get_opt(argc,argv,"f", relationFilename) ||
		!get_opt(argc, argv, "o", outputFilename))
	{
			cout<<"usage: "<<argv[0]<<" -f relationFilename -o outputFilename" << endl;
			exit(EXIT_FAILURE);
	}

	CRelation r;

	r.load(relationFilename.c_str(), true);

	global_data_t global_data;
	unsigned int size = r.getCardinality();
	SBUN * tuples = r.getBATPointer();

	global_data.tuples = tuples;

	vector<unsigned int> indices;	// vector of offsets for the start of each tuple in the file

	for (unsigned int i = 0; i < size; ++i)
	{
		indices.push_back(i);
	}

	for (unsigned int i = 0; i < r2Size; ++i)
	{
		indices.push_back(r1Size + i);
		global_data.tuples[r1Size + i].relationId = 2;
		global_data.tuples[r1Size + i].record = r2Tuples[i];
	}

	// create MapReduce scheduler args
	HMMSchedulerSpec args(argc,argv);
	args.input=&indices[0];
	args.input_size=sizeof(int)*indices.size();
	args.unit_size=sizeof(int);

	args.PrintArgs();

	// create global data as part of scheduler args
	args.global_data=&global_data;

	// store the content and keyword in a global array
	DECLARE_GLOBAL_ARRAY(tuples, (size) * sizeof(SBUN));

	// begin timing
	double t1=get_time();

	// create the map reduce scheduler
	HMMMainScheduler scheduler(args);
	
	// run map reduce
	scheduler.do_map_reduce();

	// end timing
	double t2=get_time();

	// output timing
	cout<<"Total time:     "<< t2-t1 << " seconds" << endl;

	// retrieve output
	OutputChunk output=scheduler.get_output();

	// output output
	cout<<"number of output: "<<output.count<<endl;

	ofstream fout;
	fout.open(outputFilename, std::ofstream::out | std::ofstream::binary);
	if (fout.fail())
	{
		cerr << "Error opening output file" << endl;
		exit(EXIT_FAILURE);
	}

	int joinCount = 0;
	for(int i=0;i<output.count; i++){
		unsigned int keyIdx=output.index[i].x;
		unsigned int keySize=output.index[i].y;
		unsigned int valIdx=output.index[i].z;
		unsigned int valSize = output.index[i].w;

		unsigned int key = *(int *)(output.keys + keyIdx);
		unsigned int * numOutput = (unsigned int*)(output.vals + valIdx);
		joinCount += *numOutput;
		outputTupleNode * out = (outputTupleNode *)(numOutput + sizeof(unsigned int));
		
		// cout << "Debug - no. of outputs in this array: " << *numOutput << endl;
		if (i < 10)
		{
			for (unsigned int j = 0; j < *numOutput; ++j)
			{
				//printf("(Key: %u, Value: %u, %u, %u, %u, %u, %u, Value Size: %u)\n", key, out[j].aId, out[j].bId, out[j].aKey, out[j].bKey, out[j].aData, out[j].bData, valSize);	
				printf("(Key: %u, Value: %u, %u, Value Size: %u)\n", key, out[j].aData, out[j].bData, valSize);	
			}
		}
	}

	cout << "Number of tuples joined: " << joinCount << endl;

	delete[] global_data.tuples;

	return 0;
}
